package org.chnxi.study.messagequeue.kafka.comsumer;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;
import java.util.Set;

public class SendedToMsgDetailOffsetProcessor implements OffsetProcessor<String,String> {

    private KafkaConsumer<String, String> consumer;

    @Override
    public void setKafkaConsumer(KafkaConsumer<String, String> consumer) {
        this.consumer = consumer;
    }

    @Override
    public void resetOffset() {
        // 手动指定offset
        Set<TopicPartition> topicPartitions = consumer.assignment();
        topicPartitions.forEach(partition -> {
            // TODO 从nos中获取之前消费的位置
            long newOffset = 0L;
            TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
            // 设置新的位置
            consumer.seek(topicPartition, newOffset);
        });
    }

    @Override
    public void saveOffSet(Map<TopicPartition, Long> offset) {
        offset.entrySet().forEach(entry -> {
            TopicPartition topicPartition = entry.getKey();
            Long offSet = entry.getValue();
            // todo save to nos
        });
    }
}
